package com.landcode.land.data.elasticsearch.handler;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.landcode.land.data.SynDataHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.SystemUtils;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

@Component
@Slf4j
public class EsSynDataHandler implements SynDataHandler {
    protected final static String SEP = SystemUtils.LINE_SEPARATOR;
    private final static String transaction_format = SEP + "================> binlog[{}:{}] , executeTime : {} , delay : {}ms" + SEP;
    private final static String row_format = SEP
            + "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {} , delay : {}ms"
            + SEP;

    @Override
    public void synData(Message message) throws Exception {
        //开固定数量的线程跑
        //1、创建固定线程池。
//        ExecutorService syncEntryThreadPool = Executors.newFixedThreadPool(5);
        ExecutorService syncEntryThreadPool = new ThreadPoolExecutor(5, 10, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
        int i = 0;
        List<CanalEntry.Entry> subEntries = new ArrayList<>();
        for (CanalEntry.Entry entry : message.getEntries()) {
            subEntries.add(entry);
            syncEntryThreadPool.submit(new SyncDataThread(i, subEntries));
            subEntries = new ArrayList<>();
            i++;
        }

        try {
            syncEntryThreadPool.shutdown();
            syncEntryThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
        } catch (InterruptedException e) {
            log.error("线程池超时无法释放！", e);
            throw new Exception(e);
        }
    }

    private class SyncDataThread implements Callable<Integer> {
        private int threadId;
        private List<CanalEntry.Entry> entryList;

        public SyncDataThread(int id, List<CanalEntry.Entry> entryList) {
            this.threadId = id;
            this.entryList = entryList;
        }

        @Override
        public Integer call() throws Exception {
            for (CanalEntry.Entry entry : entryList) {

                long executeTime = entry.getHeader().getExecuteTime();
                long delayTime = System.currentTimeMillis() - executeTime;

                //判断mysql事物开始和事物结束
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                    try {
                        if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN) {
                            CanalEntry.TransactionBegin begin = CanalEntry.TransactionBegin.parseFrom(entry.getStoreValue());
                            log.info(" BEGIN ----> thread id: {}", begin.getThreadId());
                        } else if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                            CanalEntry.TransactionEnd end = CanalEntry.TransactionEnd.parseFrom(entry.getStoreValue());
                            // 打印事务提交信息，事务id
                            log.info(" END ----> transaction id: {}", end.getTransactionId());
                        }
                        // 打印事务头信息，执行的线程id，事务耗时
                        log.info(transaction_format,
                                new Object[]{entry.getHeader().getLogfileName(),
                                        String.valueOf(entry.getHeader().getLogfileOffset()),
                                        String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime)});
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                }

                //解析mysql逐行数据
                if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                    CanalEntry.RowChange rowChange;
                    try {
                        rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                    } catch (Exception e) {
                        throw new RuntimeException("parse event has an error, data:" + entry.toString(), e);
                    }

                    CanalEntry.EventType eventType = rowChange.getEventType();

                    // 打印数据记录
                    log.info(row_format,
                            new Object[]{entry.getHeader().getLogfileName(),
                                    String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
                                    entry.getHeader().getTableName(), eventType,
                                    String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime)});

                    if (eventType == CanalEntry.EventType.QUERY || rowChange.getIsDdl()) {
                        log.info(" sql ----> " + rowChange.getSql() + SEP);
                    }

                    //开启线程逐行同步数据
                    if (rowChange.getRowDatasList().size() > 0) {
                        ExecutorService syncRowDataThreadPool = Executors.newFixedThreadPool(5);
                        List<CanalEntry.RowData> rowDataList = new ArrayList<>();
                        int i = 0;
                        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                            if (rowDataList.size() < 4) {
                                rowDataList.add(rowData);
                            } else {
                                rowDataList.add(rowData);//第4条加入到列表，然后利用线程池，开启线程同步
                                syncRowDataThreadPool.submit(new SyncRowDataThread(i, rowDataList, entry, eventType));
                                rowDataList = new ArrayList<>();//创建新List对象。用于开启新线程。
                            }
                        }
                        if (rowDataList.size() < 4) {//如果rowDataList中不足4条，则需要将不足的，提交到线程执行
                            syncRowDataThreadPool.submit(new SyncRowDataThread(i, rowDataList, entry, eventType));
                        }
                        try {
                            syncRowDataThreadPool.shutdown();
                            syncRowDataThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
                        } catch (Exception e) {
                            log.error("线程池超时无法释放！", e);
                            throw new Exception(e);
                        }
                    }
                }
            }

            return threadId;
        }

    }

    private class SyncRowDataThread implements Callable<Integer> {
        private int threadId;
        private List<CanalEntry.RowData> rowDataList;
        private CanalEntry.Entry entry;
        private CanalEntry.EventType eventType;

        public SyncRowDataThread(int id, List<CanalEntry.RowData> rowDataList, CanalEntry.Entry entry, CanalEntry.EventType eventType) {
            this.threadId = id;
            this.rowDataList = rowDataList;
            this.entry = entry;
            this.eventType = eventType;
        }

        @Override
        public Integer call() {
/*            Map colMap;
            for (CanalEntry.RowData rowData : rowDataList) {
                colMap = analysisColumn(rowData.getAfterColumnsList());
                if (eventType == CanalEntry.EventType.DELETE) {
                    elasticSearchDumpManager.deleteRecordToEs(colMap, entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName());
                } else {
                    serviceImportManager.getDateMap(colMap, entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName());
                }
            }
            return threadId;*/
            return 0;
        }

    }
}
